package webSocket

/*
 * @Desc: websocket
 * @author: bsyz
 * @version: v1.0.0
 */

import (
	"encoding/json"
	"fmt"
	"github.com/PhlexTheCharr/GoFurry-framework/common"
	"github.com/PhlexTheCharr/GoFurry-framework/common/log"
	"github.com/PhlexTheCharr/GoFurry-framework/common/models"
	"github.com/PhlexTheCharr/GoFurry-framework/common/service"
	"github.com/PhlexTheCharr/GoFurry-framework/roof/env"
	"github.com/gin-gonic/gin"
	"strings"
	"time"

	"github.com/gorilla/websocket"
	"github.com/sourcegraph/conc/pool"
	"net/http"
	"sync"
)

var ws = websocket.Upgrader{
	CheckOrigin: func(r *http.Request) bool {
		return true
	},
}

var heartbeat = EventMsgModel{Event: common.EVENT_HEARTBEAT}
var heartbeatSocketMsg, _ = json.Marshal(heartbeat)
var globalAdapter = &globalWsAdapter{}
var commonAdapter = &commonWsAdapter{}

// ws订阅发送线程池
var threadCache = pool.New().WithMaxGoroutines(env.GetServerConfig().Thread.WsSubscribeSendCache)

type adapter interface {
	checkAndConvert(msg string, eventFlag string) ([]byte, bool)
}

type syncConnection struct {
	Conn *websocket.Conn
	Lock *sync.Mutex
}

// 全局事件
type globalWsAdapter struct {
}

func (a *globalWsAdapter) checkAndConvert(msg string, eventFlag string) ([]byte, bool) {
	return []byte(msg), true
}

func GlobalWs(context *gin.Context) {
	subscribeAndSend(context, common.GLOBAL_MSG, false, globalAdapter)
}

// 普通事件
type commonWsAdapter struct {
}

func (a *commonWsAdapter) checkAndConvert(msg string, eventFlag string) ([]byte, bool) {
	commonMsg := models.CommonMsg{}
	err := json.Unmarshal([]byte(msg), &commonMsg)
	if err != nil {
		log.Error(fmt.Sprintf("Unmarshal err: %v", err))
		return []byte(msg), false
	}
	if strings.TrimSpace(commonMsg.Flag) == strings.TrimSpace(eventFlag) {
		return []byte(msg), true
	} else {
		return []byte(msg), false
	}
}

func CommonWs(context *gin.Context) {
	subscribeAndSend(context, common.COMMON_MSG, true, commonAdapter)
}

// 发送消息
func sendMessage(c *syncConnection, msg EventMsgModel, closeFlag *bool) {
	data, err := json.Marshal(msg)
	if err != nil {
		log.Error("Marshal err: ", err)
		return
	}
	c.Lock.Lock()
	defer c.Lock.Unlock()
	err = c.Conn.WriteMessage(1, data)
	if err != nil {
		*closeFlag = true
	}
}

func subscribeAndSend(context *gin.Context, channel string, eventFlag bool, a adapter) {
	c, err := ws.Upgrade(context.Writer, context.Request, nil)
	wsSync := &syncConnection{Conn: c, Lock: &sync.Mutex{}}
	if err != nil {
		log.Error(fmt.Sprintf("upgrade err: %v", err))
		return
	}
	// 关闭连接
	defer func() {
		_ = wsSync.Conn.Close()
		if e := recover(); e != nil {
			log.Error(fmt.Sprintf("receive subscribeAndSend recover: %v", e))
		}
	}()
	eventFlagVal := ""
	if eventFlag {
		// 订阅
		_, eventFlagReq, err := wsSync.Conn.ReadMessage()
		if err != nil {
			log.Error(fmt.Sprintf("read subscribe msg err: %v", err))
			_ = syncSend(wsSync, []byte("订阅失败"))
			return
		}
		eventFlagVal = string(eventFlagReq)
	}
	log.Info(fmt.Sprintf("receive realtime data subscribe: %v", eventFlagVal))
	// 心跳检测
	task := service.AddCronJob(30*time.Second, func() {
		defer func() {
			if e := recover(); e != nil {
				log.Error(fmt.Sprintf("receive AddCronJob recover: %v", e))
			}
		}()
		_ = syncSend(wsSync, heartbeatSocketMsg)
	})
	// 确保定时任务被清除
	defer func() {
		service.RemoveTask(task)
	}()

	closeFlag := false // 关闭标记
	// 关闭连接监听
	wsSync.Conn.SetCloseHandler(func(code int, text string) error {
		closeFlag = true
		log.Info(fmt.Sprintf("%s close :%d, %s", channel, code, text))
		return nil
	})
	errCount := 0
	// 监听处理
	go func() {
		defer func() {
			if e := recover(); e != nil {
				log.Error("receive subscribe recover: %v", e)
			}
		}()
		ch := make(chan service.EventData, env.GetServerConfig().Thread.WsStatusReceiveThread)
		service.EB.Subscribe(channel, ch)
		defer func() {
			service.EB.UnSubscribe(channel, ch)
			close(ch)
		}()
		for {
			select {
			case event := <-ch:
				threadCache.Go(func() {
					defer func() {
						if e := recover(); e != nil {
							log.Error("receive websocket recover: %v", e)
						}
					}()
					jsonMsg, e := json.Marshal(event.Data)
					if e == nil {
						sendMsg, isSend := a.checkAndConvert(string(jsonMsg), eventFlagVal)
						if isSend != false {
							se := syncSend(wsSync, sendMsg)
							if se == nil {
								log.Info("send " + channel + " msg success:" + string(sendMsg))
							} else {
								errCount++
								log.Warn("send " + channel + " msg failed:" + se.Error())
							}
						}
					}
				})
			}
			if errCount > 5 || closeFlag {
				break
			}
			// 防止空转
			time.Sleep(1 * time.Millisecond)
		}
	}()
	for {
		if closeFlag || errCount > 5 {
			log.Error("jump out")
			break
		}
		// 阻塞直到前端关闭
		_, _, _ = wsSync.Conn.ReadMessage()
		// 防止空转
		time.Sleep(1 * time.Millisecond)
	}
}

func syncSend(c *syncConnection, sendMsg []byte) error {
	c.Lock.Lock()
	defer c.Lock.Unlock()
	return c.Conn.WriteMessage(1, sendMsg)
}
